package com.I9lou.se.lurker.facet.index.support;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import org.apache.lucene.facet.taxonomy.TaxonomyReader;
import org.apache.lucene.facet.taxonomy.TaxonomyWriter;
import org.apache.lucene.facet.taxonomy.lucene.LuceneTaxonomyReader;
import org.apache.lucene.facet.taxonomy.lucene.LuceneTaxonomyWriter;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexWriterConfig.OpenMode;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.NIOFSDirectory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.I9lou.se.lurker.facet.index.TaxonomyWriterFactory;

public class SimpleTaxonomyWriterFactory implements TaxonomyWriterFactory {

	private static final Logger LOG = LoggerFactory.getLogger(SimpleTaxonomyWriterFactory.class);
	
	private ConcurrentHashMap<Integer,String> facetIndexPath = new ConcurrentHashMap<Integer,String>();
	
	private ConcurrentHashMap<Integer,Directory> taxonomyDirectory = new ConcurrentHashMap<Integer, Directory>();
	
	private ConcurrentHashMap<Integer,TaxonomyWriter> taxonomyWriter = new ConcurrentHashMap<Integer, TaxonomyWriter>();
	
	/**
	 * 用于监控外部调度器的锁
	 * 只有当indexWriterLatch计数完毕，才能关闭IndexWriterFactory
	 */
	private CountDownLatch taxonomyWriterLatch;
	
	/** 上一轮使用的聚类reader引用暂存一个周期 */
	private ConcurrentHashMap<Integer, TaxonomyReader> lastTaxonomyReaderPool = new ConcurrentHashMap<Integer, TaxonomyReader>();

	/** 当前正在使用聚类reader容器 */
	private ConcurrentHashMap<Integer, TaxonomyReader> currentTaxonomyReaderPool = new ConcurrentHashMap<Integer, TaxonomyReader>();

	public void setFacetIndexPath(HashMap<Integer, String> facetIndexPath) {
		this.facetIndexPath.putAll(facetIndexPath);
	}

	@Override
	public CountDownLatch getTaxonomyWriterLatch() {
		return taxonomyWriterLatch;
	}

	public void init(){
		//初始化taxonomyWriterLatch调度器的关闭锁
		taxonomyWriterLatch = new CountDownLatch(1);
		setupTaxonomyWriter();
		recreateTaxonomyReader();
	}
	
	@Override
	public void destroy() {
		try {
			//等待调度器关闭
			taxonomyWriterLatch.await();
		} catch (InterruptedException e1) {} //throw it away
		try {
			long start = System.currentTimeMillis();
			for (TaxonomyWriter writer : taxonomyWriter.values()) {
				writer.close();
			}
			for(TaxonomyReader reader : lastTaxonomyReaderPool.values()){
				reader.close();
			}
			for(TaxonomyReader reader : currentTaxonomyReaderPool.values()){
				reader.close();
			}
			LOG.info("[FacetIndex-Shutdown] facetIndexPath: \"{}\"; cost: {}ms", this.facetIndexPath, (System.currentTimeMillis()-start));
		} catch (IOException e) {
			LOG.error("Close TaxonomyWriter faild", e);
		}
	}
	
	private void setupTaxonomyWriter() {
		try {
			for (int key: facetIndexPath.keySet()){
				taxonomyDirectory.put(key, new NIOFSDirectory(new File(facetIndexPath.get(key))));
				taxonomyWriter.put(key, new LuceneTaxonomyWriter(taxonomyDirectory.get(key), OpenMode.CREATE_OR_APPEND));
			}
			LOG.info("[TaxonomyWriter-Create]: FacetIndexPath:{}", facetIndexPath);
		} catch (IOException e) {
			LOG.error("Create TaxonomyWriter faild! Direactory: {};", facetIndexPath, e);
		}
	}

	@Override
	public Directory getTaxonomyDirectory(int appId) {
		return taxonomyDirectory.get(appId);
	}

	@Override
	public Map<Integer,TaxonomyWriter> getTaxonomyWriter() {
		return taxonomyWriter;
	}
	
	/**
	 * 现在只有个路径用来存放聚类，所以暂不考虑通过facetIndexPath来获取
	 */
	@Override
	public TaxonomyReader getTaxonomyReader(int appId) {
		return currentTaxonomyReaderPool.get(appId);
	}

	@Override
	public String toString() {
		return "SimpleTaxonomyWriterFactory [taxonomyDirectory="
				+ taxonomyDirectory + "]";
	}
	
	public void recreateTaxonomyReader(){
		try {
			for (int key: facetIndexPath.keySet()){
				TaxonomyReader oldTaxonomyReader =
						currentTaxonomyReaderPool.put(key, new LuceneTaxonomyReader(taxonomyDirectory.get(key)));
				if(oldTaxonomyReader != null){
					TaxonomyReader grandPaTaxonomyReader = lastTaxonomyReaderPool.put(key, oldTaxonomyReader);
					if(grandPaTaxonomyReader != null){
						try {
							grandPaTaxonomyReader.close();
						} catch (IOException e) {
							LOG.error("Error when closing "+grandPaTaxonomyReader, e);
						}
					}
				}
			}
		} catch (CorruptIndexException e) {
			LOG.error("Error when creating reader from path:"+facetIndexPath, e);
		} catch (IOException e) {
			LOG.error("Error when creating reader from path:"+facetIndexPath, e);
		}
	}
	
}
